/*
 * Copyright 2017-2020 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.gitee.minimalismstyle.fresh.common.data.jdbc.beetl

import org.springframework.data.relational.core.sql.IdentifierProcessing
import org.beetl.sql.core.SQLManager
import org.beetl.sql.core.SqlId
import org.slf4j.LoggerFactory
import org.springframework.data.mapping.PersistentPropertyPath
import org.springframework.data.relational.core.mapping.RelationalPersistentProperty
import org.springframework.dao.EmptyResultDataAccessException
import org.springframework.data.domain.Pageable
import org.springframework.data.domain.Sort
import org.springframework.data.jdbc.core.convert.*
import org.springframework.data.relational.core.dialect.Dialect
import org.springframework.data.relational.core.sql.SqlIdentifier
import java.util.stream.Collectors
import org.springframework.data.relational.core.mapping.RelationalMappingContext
import org.springframework.data.relational.core.sql.LockMode
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations
import org.springframework.util.Assert
import java.util.*
import java.util.function.Function

/**
 * [DataAccessStrategy] implementation based on MyBatis. Each method gets mapped to a statement. The name of the
 * statement gets constructed as follows: By default, the namespace is based on the class of the entity plus the suffix
 * "Mapper". This is then followed by the method name separated by a dot. For methods taking a [PropertyPath] as
 * argument, the relevant entity is that of the root of the path, and the path itself gets as dot separated String
 * appended to the statement name. Each statement gets an instance of [BeetlContext], which at least has the
 * entityType set. For methods taking a [PropertyPath] the entityType if the context is set to the class of the
 * leaf type.
 *
 * @author Jens Schauder
 * @author Kazuki Shimizu
 * @author Oliver Gierke
 * @author Mark Paluch
 * @author Tyler Van Gorder
 * @author Milan Milanov
 * @author Myeonghyeon Lee
 */
class BeetlDataAccessStrategy(val sqlManager: SQLManager, val identifierProcessing: IdentifierProcessing) : DataAccessStrategy {

    private var namespaceStrategy = NamespaceStrategy.DEFAULT_INSTANCE

    /**
     * Set a NamespaceStrategy to be used.
     *
     * @param namespaceStrategy Must be non null
     */
    fun setNamespaceStrategy(namespaceStrategy: NamespaceStrategy) {
        Assert.notNull(namespaceStrategy, "The NamespaceStrategy must not be null")
        this.namespaceStrategy = namespaceStrategy
    }

    /*
	 * (non-Javadoc)
	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#insert(java.lang.Object, java.lang.Class, ParentKeys)
	 */
    override fun <T> insert(instance: T, domainType: Class<T>, identifier: Identifier): Any? {
        val beetlContext = BeetlContext(identifier, instance, domainType)
        val sqlId = SqlId.of(namespace(domainType), "insert")
        sqlManager().insert(sqlId, instance)
        return beetlContext.id
    }

    /*
	 * (non-Javadoc)
	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#update(java.lang.Object, java.lang.Class)
	 */
    override fun <S> update(instance: S, domainType: Class<S>): Boolean {
        val sqlId = SqlId.of(namespace(domainType), "update")
        return sqlManager().update(sqlId,instance) !== 0
    }

    /*
	 * (non-Javadoc)
	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#updateWithVersion(java.lang.Object, java.lang.Class, java.lang.Number)
	 */
    override fun <S> updateWithVersion(instance: S, domainType: Class<S>, previousVersion: Number): Boolean {
        val sqlId = SqlId.of(namespace(domainType), "updateWithVersion")
        return sqlManager().update(sqlId, instance) !== 0
    }


    /*
	 * (non-Javadoc)
	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#delete(java.lang.Object, java.lang.Class)
	 */
    override fun delete(id: Any, domainType: Class<*>) {
        val sqlId = SqlId.of(namespace(domainType), "delete")
        sqlManager().update(sqlId, id)
    }

    /*
	 * (non-Javadoc)
	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#deleteInstance(java.lang.Object, java.lang.Class)
	 */
    override fun <T> deleteWithVersion(id: Any, domainType: Class<T>, previousVersion: Number) {
        val sqlId = SqlId.of(namespace(domainType), "deleteWithVersion")
        sqlManager().update(sqlId, id)
    }

    /*
	 * (non-Javadoc)
	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#delete(java.lang.Object, org.springframework.data.mapping.PersistentPropertyPath)
	 */
    override fun delete(rootId: Any, propertyPath: PersistentPropertyPath<RelationalPersistentProperty>) {
        val ownerType = propertyPath.baseProperty!!.owner.type
        val sqlId = SqlId.of(namespace(ownerType), ".delete-" + toDashPath(propertyPath))
        val leafType = propertyPath.requiredLeafProperty.typeInformation.type
        val parameter = BeetlContext(rootId, null, leafType, emptyMap())
        sqlManager().update(sqlId, parameter)
    }

    /*
	 * (non-Javadoc)
	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#deleteAll(java.lang.Class)
	 */
    override fun <T> deleteAll(domainType: Class<T>) {
        val sqlId = SqlId.of(namespace(domainType), "deleteAll")
        val parameter = BeetlContext(null, null, domainType, emptyMap())
        sqlManager().update(sqlId, parameter)
    }

    /*
	 * (non-Javadoc)
	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#deleteAll(org.springframework.data.mapping.PersistentPropertyPath)
	 */
    override fun deleteAll(propertyPath: PersistentPropertyPath<RelationalPersistentProperty>) {
        val baseType = propertyPath.baseProperty!!.owner.type
        val leafType = propertyPath.requiredLeafProperty.typeInformation.type
        val sqlId = SqlId.of(namespace(baseType), ".deleteAll-" + toDashPath(propertyPath))
        val parameter = BeetlContext(null, null, leafType, emptyMap())
        sqlManager().update(sqlId, parameter)
    }

    /*
	 * (non-Javadoc)
	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#acquireLockById(java.lang.Object, org.springframework.data.relational.core.sql.LockMode, java.lang.Class)
	 */
    override fun <T> acquireLockById(id: Any, lockMode: LockMode, domainType: Class<T>) {
        val sqlId = SqlId.of(namespace(domainType), "acquireLockById")
        val parameter = BeetlContext(id, null, domainType, emptyMap())
        val result: Long = sqlManager().selectSingle(sqlId, parameter, Long::class.java)
        if (result < 1) {
            val message = String.format("The lock target does not exist. id: %s, statement: %s", id, sqlId)
            throw EmptyResultDataAccessException(message, 1)
        }
    }

    /*
	 * (non-Javadoc)
	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#acquireLockAll(org.springframework.data.relational.core.sql.LockMode, java.lang.Class)
	 */
    override fun <T> acquireLockAll(lockMode: LockMode, domainType: Class<T>) {
        val sqlId = SqlId.of(namespace(domainType), "acquireLockAll")
        val parameter = BeetlContext(null, null, domainType, emptyMap())
        sqlManager().selectSingle(sqlId, parameter, domainType)
    }

    /*
	 * (non-Javadoc)
	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#findById(java.lang.Object, java.lang.Class)
	 */
    override fun <T> findById(id: Any, domainType: Class<T>): T? {
        val sqlId = SqlId.of(namespace(domainType), "findById")
        val parameter = BeetlContext(id, null, domainType, emptyMap())
        return sqlManager().selectSingle(sqlId, parameter, domainType)
    }

    /*
	 * (non-Javadoc)
	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#findAll(java.lang.Class)
	 */
    override fun <T> findAll(domainType: Class<T>): Iterable<T> {
        val sqlId = SqlId.of(namespace(domainType), "findAll")
        return sqlManager().select(sqlId, domainType)
    }

    /*
	 * (non-Javadoc)
	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#findAllById(java.lang.Iterable, java.lang.Class)
	 */
    override fun <T> findAllById(ids: MutableIterable<*>, domainType: Class<T>): MutableIterable<T> {
        val sqlId = SqlId.of(namespace(domainType), "findAllById")
        return sqlManager().select(sqlId, domainType,
                BeetlContext(ids, null, domainType, emptyMap()))
    }

    override fun findAllByPath(identifier: Identifier,
                               path: PersistentPropertyPath<out RelationalPersistentProperty>): Iterable<Any> {
        val sqlId = SqlId.of(namespace(path.baseProperty!!.owner.type), "findAllByPath-" + path.toDotPath())
        return sqlManager().select(sqlId, path.baseProperty!!.owner.type,
                BeetlContext(identifier, null, path.requiredLeafProperty.type))
    }

    /*
	 * (non-Javadoc)
	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#existsById(java.lang.Object, java.lang.Class)
	 */
    override fun <T> existsById(id: Any, domainType: Class<T>): Boolean {
        val sqlId = SqlId.of(namespace(domainType), "existsById")
        val parameter = BeetlContext(id, null, domainType, emptyMap())
        return sqlManager().selectSingle(sqlId, parameter, Int::class.java) != 0
    }

    /*
	 * (non-Javadoc)
	 * @see org.springframework.data.jdbc.core.JdbcAggregateOperations#findAll(java.lang.Class, org.springframework.data.domain.Sort)
	 */
    override fun <T> findAll(domainType: Class<T>, sort: Sort): Iterable<T> {
        val additionalContext: MutableMap<String, Any> = HashMap()
        additionalContext["sort"] = sort
        val sqlId = SqlId.of(namespace(domainType), "findAllSorted")
        return sqlManager().select(sqlId, domainType,
                BeetlContext(null, null, domainType, additionalContext))
    }

    /*
	 * (non-Javadoc)
	 * @see org.springframework.data.jdbc.core.JdbcAggregateOperations#findAll(java.lang.Class, org.springframework.data.domain.Pageable)
	 */
    override fun <T> findAll(domainType: Class<T>, pageable: Pageable): Iterable<T> {
        val additionalContext: MutableMap<String, Any> = HashMap()
        additionalContext["pageable"] = pageable
        val sqlId = SqlId.of(namespace(domainType), "findAllPaged")
        return sqlManager().select(sqlId, domainType,
                BeetlContext(null, null, domainType, additionalContext))
    }

    /*
	 * (non-Javadoc)
	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#count(java.lang.Class)
	 */
    override fun count(domainType: Class<*>): Long {
        val parameter = BeetlContext(null, null, domainType, emptyMap())
        val sqlId = SqlId.of(namespace(domainType), "count")
        return sqlManager().selectSingle(sqlId, parameter, Long::class.java)
    }

    private fun convertToParameterMap(additionalParameters: Map<SqlIdentifier, Any>): Map<String, Any> {
        val params = mutableMapOf<String, Any>()
        additionalParameters.forEach { (k, v) -> params[k.toSql(identifierProcessing)] = v}
        return params
    }

    private fun namespace(domainType: Class<*>): String {
        return namespaceStrategy.getNamespace(domainType)
    }

    private fun sqlManager(): SQLManager {
        return sqlManager
    }

    companion object {
        private val LOG = LoggerFactory.getLogger(BeetlDataAccessStrategy::class.java)
        private const val VERSION_SQL_PARAMETER_NAME_OLD = "___oldOptimisticLockingVersion"

        /**
         * Create a [DataAccessStrategy] that first checks for queries defined by MyBatis and if it doesn't find one
         * uses a [DefaultDataAccessStrategy]
         */
        fun createCombinedAccessStrategy(context: RelationalMappingContext?,
                                         converter: JdbcConverter?, operations: NamedParameterJdbcOperations?, sqlManager: SQLManager, dialect: Dialect): DataAccessStrategy {
            return createCombinedAccessStrategy(context, converter, operations, sqlManager, NamespaceStrategy.DEFAULT_INSTANCE,
                    dialect)
        }

        /**
         * Create a [DataAccessStrategy] that first checks for queries defined by MyBatis and if it doesn't find one
         * uses a [DefaultDataAccessStrategy]
         */
        fun createCombinedAccessStrategy(context: RelationalMappingContext?,
                                         converter: JdbcConverter?, operations: NamedParameterJdbcOperations?, sqlManager: SQLManager,
                                         namespaceStrategy: NamespaceStrategy, dialect: Dialect): DataAccessStrategy {

            // the DefaultDataAccessStrategy needs a reference to the returned DataAccessStrategy. This creates a dependency
            // cycle. In order to create it, we need something that allows to defer closing the cycle until all the elements are
            // created. That is the purpose of the DelegatingAccessStrategy.
            val delegatingDataAccessStrategy = DelegatingDataAccessStrategy()
            val beetlDataAccessStrategy = BeetlDataAccessStrategy(sqlManager,
                    dialect.identifierProcessing)
            beetlDataAccessStrategy.setNamespaceStrategy(namespaceStrategy)
            val cascadingDataAccessStrategy = CascadingDataAccessStrategy(
                    Arrays.asList(beetlDataAccessStrategy, delegatingDataAccessStrategy))
            val sqlGeneratorSource = SqlGeneratorSource(context!!, converter!!, dialect)
            val defaultDataAccessStrategy = DefaultDataAccessStrategy( //
                    sqlGeneratorSource,  //
                    context,  //
                    converter,  //
                    operations!! //
            )
            delegatingDataAccessStrategy.setDelegate(defaultDataAccessStrategy)
            return cascadingDataAccessStrategy
        }

        private fun toDashPath(propertyPath: PersistentPropertyPath<RelationalPersistentProperty>): String {
            return propertyPath.toDotPath()!!.replace("\\.".toRegex(), "-")
        }
    }


}